package com.nulldev.util.threading.fastES;

import java.util.Collection;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedTransferQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import com.nulldev.util.threading.ThreadUtils;

public class LtqExecutor<T> implements ExecutorService {
	private volatile boolean running;
    private volatile boolean stopped;
    private final int threadCount;
    private final BlockingQueue<Runnable> queue;
    private final Thread[] threads;

    /**
     * Default Value: 128 threads
     */
    public LtqExecutor() {
    	this(128);
    }
    
    public LtqExecutor(final int threadCount) {
        this.threadCount = threadCount;
        this.queue = new LinkedTransferQueue<Runnable>();
        this.running = true;
        this.stopped = false;
        this.threads = new Thread[threadCount];
        for(int i = 0; i < threadCount; i++) {
        	this.threads[i] = new Thread(new Worker());
        	this.threads[i].start();
        }
    }

    @Override
    public void execute(Runnable runnable) {
        try {
        	this.queue.put(runnable);
        } catch (InterruptedException ie) {
            throw new RuntimeException(ie);
        }
    }

    private class Worker implements Runnable {
    	@Override
        public void run() {
            while(running) {
                Runnable runnable = null;
                try {
                    runnable = queue.take();
                } catch (InterruptedException ie) {
                    // was interrupted - just go round the loop again,
                    // if running = false will cause it to exit
                }
                try {
                    if (runnable != null) { 
                        runnable.run();
                    }
                } catch (Exception e) {
                    System.out.println("failed because of: " + e.toString());
                    e.printStackTrace();
                }
            }
        }
    }

	@Override
	public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
		this.queue.clear();
		ThreadUtils.kill(this.threads);
		this.running = false;
		return this.stopped = true;
	}

	@Override
	public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException {
		throw new UnsupportedOperationException("invokeAll(tasks) -> Not implemented!");
	}

	@Override
	public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
			throws InterruptedException {
		throw new UnsupportedOperationException("invokeAll(tasks, timeout, unit) -> Not implemented!");
	}

	@Override
	public <T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException {
		throw new UnsupportedOperationException("invokeAny(tasks) -> Not implemented!");
	}

	@Override
	public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
			throws InterruptedException, ExecutionException, TimeoutException {
		throw new UnsupportedOperationException("invokeAny(tasks, timeout, unit) -> Not implemented!");
	}

	@Override
	public boolean isShutdown() {
		return this.stopped;
	}

	@Override
	public boolean isTerminated() {
		return this.stopped;
	}

	@Override
	public void shutdown() {
		throw new UnsupportedOperationException("shutdown() -> Not implemented!");
	}

	@Override
	public List<Runnable> shutdownNow() {
		throw new UnsupportedOperationException("shutdownNow() -> Not implemented!");
	}

	@Override
	public <T> Future<T> submit(Callable<T> task) {
		throw new UnsupportedOperationException("submit(task) -> Not implemented!");
	}

	@Override
	public Future<?> submit(Runnable task) {
		this.execute(task);
		return null;
	}

	@Override
	public <T> Future<T> submit(Runnable task, T result) {
		throw new UnsupportedOperationException("submit(task, T) -> Not implemented!");
	}
	
	public int getThreadCount() {
		return this.threadCount;
	}
}
